package com.hopu.user.total;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.total.Total;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class TotalAna {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<Total, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String type = json.getString("event_type");

            return new Tuple2<>(new Total(type), 1);
        });

        //过滤出三个不同类型的RDD
        JavaPairRDD<Total, Integer> viewRDD = pairRDD.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<Total, Integer> cartRDD = pairRDD.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<Total, Integer> parchaseRDD = pairRDD.filter(t -> "parchase".equals(t._1.getType()));

        //分别统计三个不同类型的RDD的总量
        Total view = getTotal(viewRDD);
        Total cart = getTotal(cartRDD);
        Total parchase = getTotal(parchaseRDD);

        //合并
        view.setCart(cart.getCart());
        view.setParchase(parchase.getParchase());
        view.setTotal(view.getView()+view.getCart()+view.getParchase());

        System.out.println(view);
        //写入到HBase
        String[] columns = {"view", "cart", "parchase", "total"};
        String[] values = {view.getView()+"", view.getCart()+"", view.getParchase()+"",
            view.getTotal()+""};

        try {
            HBaseUtils.putOneRowToHbase("totalana", "1001", "totalinfo", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Total getTotal(JavaPairRDD<Total, Integer> rdd) {
        JavaPairRDD<Total, Iterable<Integer>> group = rdd.groupByKey();

        JavaRDD<Total> map = group.map(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }

            if ("view".equals(t._1.getType())) {
                t._1.setView(count);
            } else if ("cart".equals(t._1.getType())) {
                t._1.setCart(count);
            } else {
                t._1.setParchase(count);
            }
            return t._1;
        });
        List<Total> take = map.take(1);
        return take.get(0);
    }
}
